package storm.online.bolt;

import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

public class LogBehaviorPrinterBolt extends BaseRichBolt {

	private OutputCollector collector;
	
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collector = collector;
	}

	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		
		String gameId = tuple.getStringByField("gameId");
		long areaId = tuple.getLongByField("areaId");
		int numChange = tuple.getIntegerByField("numChange");
		
		System.out.println("game_id is " + gameId + ", area_id is " + areaId + ", numberChange is " + numChange );
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare( new Fields("gameId", "areaId", "groupId", "numChange"));
	}

}
